from pyhive import hive
import pandas as pd
from sqlalchemy import create_engine

engine = create_engine('mysql+pymysql://root:123456@192.168.88.161:3306/book_movies_olpk')

if __name__ == '__main__':
    # 获取到Hive(Spark ThriftServer的链接)
    conn = hive.Connection(host="node1", port=10000, username="root", database='book_movies_olpk')
    cursor = conn.cursor()

    # Task-1 价格从高到低快速分析
    cursor.execute("select name, price from bookinfo where price > 2 order by price limit 20;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['name', 'price']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test1', engine, index=False)
    print("Task—1 finished")

    # Task-2 评分表
    cursor.execute("select name, rating from bookinfo where rating >= 90 order by rating desc; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['name', 'rating']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test2', engine, index=False)
    print("Task—2 finished")

    # Task-3 推荐分进行排序
    cursor.execute("select name, Recommend, publisher from bookinfo order by Recommend desc; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['name', 'recommend', 'publisher']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test3', engine, index=False)
    print("Task—3 finished")

    # Task-4 不同出版社每年推出的图书有多少
    cursor.execute(
        "select substr(publish_date,1,4), publisher, count(name) from bookinfo group by publisher, substr(publish_date,1,4) having count(name) >= 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['publish_date', 'publisher', 'name_count']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test4', engine, index=False)
    print("Task—4 finished")

    # Task-5 每种图书类型每年推出多少本
    cursor.execute(
        "select substr(publish_date,1,4), tag, count(name) from bookinfo group by tag, substr(publish_date,1,4) having count(name) >= 25    and tag != \"总榜\";  ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['publish_date', 'tag', 'name_count']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test5', engine, index=False)
    print("Task—5 finished")

    # Task-6 每个作者图书的好评数总计排名前20
    cursor.execute(
        "select author, sum(rating) as like_name from bookinfo where length(author) != 0 group by author order by like_name desc limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['author', 'count_rating']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test6', engine, index=False)
    print("Task—6 finished")

    # Task-7 推荐分90以上的作者数据作者排名
    cursor.execute(
        "select author, Recommend from bookinfo where Recommend > 90   and length(author) != 0 order by Recommend desc limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['author', 'recommend']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test7', engine, index=False)
    print("Task—7 finished")

    # Task-8 作者图书在数据里的占比
    cursor.execute(
        "select author, count(1) as like_me from bookinfo where length(author) != 0 group by author order by like_me desc limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['author', 'like_name']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test8', engine, index=False)
    print("Task—8 finished")

    # Task-9 书籍总评排名
    cursor.execute("select name, discuss from bookinfo order by discuss desc limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['name', 'discuss']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test9', engine, index=False)
    print("Task—9 finished")

    # Task-10 每种图书类型的总计
    cursor.execute("select tag, count(1) as tag_me from bookinfo group by tag order by tag_me desc limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['tag', 'tag_me']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test10', engine, index=False)
    print("Task—10 finished")

    # Task-11 日均访问量，日均用户量
    cursor.execute("select cast(datetime as date) as day,sum(case when behavior_type = 'pv' then 1 else 0 end) as pv,count(distinct user_id) as uv "
                   "from user_behavior "
                   "group by cast(datetime as date)"
                   "order by day "
                   "limit 20; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['day', 'pv','uv']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test11', engine, index=False)
    print("Task—11 finished")

    # Task-12 每个用户的购物情况
    cursor.execute("select user_id,"
                   "sum(case when behavior_type = 'pv' then 1 else 0 end) as pv, "   #点击数
                   "sum(case when behavior_type = 'fav' then 1 else 0 end) as fav, "  #收藏数
                   "sum(case when behavior_type = 'cart' then 1 else 0 end) as cart, "  #加购物车数
                   "sum(case when behavior_type = 'buy' then 1 else 0 end) as buy "  #购买数
                   "from user_behavior "
                   "group by user_id limit 20;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['user_id', 'pv','fav','cart','buy']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test12', engine, index=False)
    print("Task—12 finished")

    # Task-13 产生两次或两次以上购买的用户占购买用户的比例
    cursor.execute("select sum(case when buy > 1 then 1 else 0 end) as sum, "
                   "sum(case when buy > 1 then 1 else 0 end) / sum(case when buy > 0 then 1 else 0 end) as rate "
                   "from user_behavior_count; ")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['sum','rate']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test13', engine, index=False)
    print("Task—13 finished")

    # Task-14 一天的活跃时段分布
    cursor.execute("select hour(datetime) as hour, "
                   "sum(case when behavior_type = 'pv' then 1 else 0 end) as pv, "   #点击数
                   "sum(case when behavior_type = 'fav' then 1 else 0 end) as fav, "  #收藏数
                   "sum(case when behavior_type = 'cart' then 1 else 0 end) as cart, "  #加购物车数
                   "sum(case when behavior_type = 'buy' then 1 else 0 end) as buy  " #购买数
                   "from user_behavior "
                   "group by hour(datetime) "
                   "order by hour;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['hour', 'pv','fav','cart','buy']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test14', engine, index=False)
    print("Task—14 finished")

    # Task-15 根据消费频率判断最忠实用户
    cursor.execute("select user_id,count(1) as F,dense_rank() over(order by count(1) desc) as F_rank "
                   "from user_behavior "
                   "where behavior_type = 'buy' "
                   "group by user_id "
                   "limit 10;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    # list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns = ['user_id', 'F','F_rank']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('bookinfo_test15', engine, index=False)
    print("Task—15 finished")
